package com.rocketmq.transaction.listener;

import com.rocketmq.exception.BusinessImplNullException;
import com.rocketmq.transaction.service.TransactionService;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.producer.LocalTransactionState;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;

/**
 * 事务执行器
 *     主要用于执行本地事务，并提交本地事务执行结果
 * 若本地事务长时间没有响应
 *     提供事务回查方法
 * @author: hw
 * @date: 2021/1/12 10:46
 */

@Slf4j
public class DefaultTransactionListener implements TransactionListener {

    private TransactionService transactionService;

    public DefaultTransactionListener(TransactionService transactionService){
        this.transactionService = transactionService;
    }


    /**
     * 当send transactional prepare(half)消息成功时，将调用此方法来执行本地事务。
     * @param message  一半的消息
     * @param o  自定义业务参数
     * @return  事务状态
     */
    @Override
    public LocalTransactionState executeLocalTransaction(Message message, Object o) {
        if (null == transactionService){
            throw new BusinessImplNullException("transaction concrete realization is not null");
        }
        try {
            transactionService.sendTransaction(message.getTransactionId());
            return LocalTransactionState.COMMIT_MESSAGE;
        }catch (BusinessImplNullException e){
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }catch (Exception e){
            e.printStackTrace();
            return LocalTransactionState.ROLLBACK_MESSAGE;
        }
    }

    @Override
    public LocalTransactionState checkLocalTransaction(MessageExt messageExt){
        /**
         *  查询数据库此消息是否已处理
         *
         */
        if (transactionService.transactionCallBack(messageExt.getTransactionId())){
            return LocalTransactionState.COMMIT_MESSAGE;
        }
        return LocalTransactionState.UNKNOW;
    }

}
